package com.troila.rpc;

import com.troila.client.grpc.core.DataRpcServiceGrpc;
import com.troila.client.grpc.core.MessageChannelRequest;
import com.troila.client.grpc.core.ResourcePageRequest;
import com.troila.client.grpc.core.ResourcePageResponse;
import com.troila.config.RpcConfig;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class RpcClient {

    private RpcConfig config;

    private MessageStreamListener listener;

    @Autowired
    public void setConfig(RpcConfig config) {
        this.config = config;
    }

    @Autowired
    public void setListener(MessageStreamListener listener) {
        this.listener = listener;
    }

    public void listener() throws InterruptedException {
        ManagedChannel channel = NettyChannelBuilder
                .forAddress(config.getClient().getServerHost(), config.getClient().getServerPort())
                .usePlaintext()
                .intercept(new MetaDataInterceptor(UUID.randomUUID().toString()))
                .build();
        DataRpcServiceGrpc.DataRpcServiceStub stub = DataRpcServiceGrpc.newStub(channel);
        log.info("rpc client started on server host : {} | port: {}", config.getClient().getServerHost(), config.getClient().getServerPort());
        stub.messageChannel(MessageChannelRequest.newBuilder().build(), listener);
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
    }

    public void query(ResourcePageRequest request) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder
                .forAddress(config.getClient().getServerHost(), config.getClient().getServerPort())
                .usePlaintext()
                .intercept(new MetaDataInterceptor(UUID.randomUUID().toString()))
                .build();
        DataRpcServiceGrpc.DataRpcServiceBlockingStub stub = DataRpcServiceGrpc.newBlockingStub(channel);
        ResourcePageResponse resourcePage = stub.findResourcePage(request);
        resourcePage.getResourceList().stream()
                .forEach(resource -> log.info("id is {}, name is {}", resource.getId(), resource.getName()));
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);

    }

}
